package com.jfzt.utils;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.alibaba.fastjson2.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author: shunpeng.hu
 * @date: 2024/3/6 10:26
 */
@Slf4j
public class ElasticsearchServiceImplUtils {

    @Resource
    private RestHighLevelClient client;
    @Resource
    private ElasticsearchClient elasticsearchClient;

    private static RestHighLevelClient restHighLevelClient;
    private static ElasticsearchClient ec;

    static {
        // 配置es用户名、密码、连接地址
//        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(
//                RestClient.builder(new HttpHost("localhost", 9200, "http"))
//                        .setDefaultHeaders(new Header[]{new BasicHeader("Authorization",
//                                "Basic " + Base64.getEncoder().encodeToString("elastic:123456".getBytes()))})
//        );
//        ElasticsearchServiceImplUtils.restHighLevelClient = restHighLevelClient;

        /** ============================================================== */
        // 配置es请求头
        Header[] defaultHeaders = new Header[]{new BasicHeader("X-Elastic-Product", "Elasticsearch")};
        RestClient restClient = RestClient.builder(
                        new HttpHost("localhost", 9200)
                )
                .setHttpClientConfigCallback(httpAsyncClientBuilder -> {
                    //账密设置
                    CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                    //es账号密码（一般使用,用户elastic）
                    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("elastic", "123456"));
                    httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);

                    return httpAsyncClientBuilder;
                })
                .setDefaultHeaders(defaultHeaders)
                .build();

//        restHighLevelClient= new RestHighLevelClient(restClient);

        RestClientTransport restClientTransport = new RestClientTransport(restClient, new JacksonJsonpMapper());

        ec = new ElasticsearchClient(restClientTransport);
        /** ============================================================== */

    }

    @PostConstruct
    public void elasticsearchServiceImpl() {
        restHighLevelClient = this.client;
//        ec = this.elasticsearchClient;
    }

    // 线程池大小
    private static final int THREAD_POOL_SIZE = 10;

    public static final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

    // 创建阻塞队列
    private static final BlockingQueue<Runnable> TASK_QUEUE = new LinkedBlockingQueue<>();

    // 添加任务到队列
    public static void addTaskToQueue(String userIndexName, Map<String, Object> document) {
        TASK_QUEUE.add(() -> createUserDocument(userIndexName, document));
    }

    public static void createUserDocument(String userIndexName, Map<String, Object> document) {
        try {
            SearchRequest searchRequest = new SearchRequest(userIndexName);

            searchRequest.source().query(QueryBuilders.matchAllQuery()).trackTotalHits(true);
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            Long value = response.getHits().getTotalHits().value;
            Long _id = (value == 0) ? value = 1L : value + 1L;

            //创建Request对象
            IndexRequest indexRequest = new IndexRequest(userIndexName).id(String.valueOf(_id)).source(document);

            //发送请求
            restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new ElasticsearchStatusException("创建用户索引文档异常~", RestStatus.fromCode(500), e.getMessage());
        }
    }

    public static boolean isExistsIndex(String indexName) {
        boolean isExists;
        try {
//            GetRequest request = new GetRequest.Builder().index(indexName).build();
            org.elasticsearch.client.indices.GetIndexRequest request = new GetIndexRequest(indexName);
//            ExistsRequest request = new ExistsRequest.Builder().index(indexName).build();
//            BooleanResponse exists = ec.indices().exists(request);
//            System.out.println(exists);
//            IndexRequest request = new IndexRequest(indexName);
            isExists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
//            System.out.println(isExists);
        } catch (IOException e) {
            throw new ElasticsearchStatusException("索引异常~", RestStatus.fromCode(500), e.getMessage());
        }
        return null == null;
    }

    /**
     * 检查索引是否存在
     * @param indexName
     * @return false存在；true：不存在
     */
    public static boolean checkIndexExists(String indexName) {
        try {
//        IndicesExistsRequest request = new IndicesExistsRequest(indexName);
            GetIndexRequest request = new GetIndexRequest(indexName);

            boolean response = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
            return response;
        } catch (Exception e) {
            log.error("xxexist");
            return false;
        }
    }

    /**
     * 创建登录索引库 indexName
     *
     * @param indexName 索引库
     */
    public static void createIndex(String indexName) {
        try {
            // 创建Request对象
            CreateIndexRequest request = new CreateIndexRequest(indexName);

            //发送请求
            restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new ElasticsearchStatusException("创建索引库异常~", RestStatus.fromCode(500), e.getMessage());
        }
    }

    /**
     * 创建登录索引文档
     *
     * @param indexName 索引
     * @param document  文档
     */
    public static void createIndexDocument(String indexName, Map<String, Object> document) {
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);

            searchRequest.source().query(QueryBuilders.matchAllQuery()).trackTotalHits(true);
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            long value = response.getHits().getTotalHits().value;
            long _id = (value == 0) ? value = 1 : value + 1;

            //创建Request对象
            IndexRequest indexRequest = new IndexRequest(indexName).id(String.valueOf(_id)).source(document);

            //发送请求
            restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new ElasticsearchStatusException("创建索引文档异常~", RestStatus.fromCode(500), e.getMessage());
        }
    }

    public static void commonWriteElasticSearch(HttpServletRequest request, JSONObject esJsonObject, HttpServletResponse response) {
        // 建立ES索引文档
        if (!ElasticsearchServiceImplUtils.isExistsIndex("keycloak_util_dev")) {
            // 不存在索引库，创建索引库
            ElasticsearchServiceImplUtils.createIndex("keycloak_util_dev");
        }
//        HttpRequestInterceptior.getHttpServletRequest(request, esJsonObject, response);
        ElasticsearchServiceImplUtils.createIndexDocument("keycloak_util_dev", esJsonObject);
    }

    // 从队列中获取任务并执行
    public static void executeTasks() {
        while (!TASK_QUEUE.isEmpty()) {
            executorService.execute(TASK_QUEUE.poll());
        }
    }


}
